// Copyright 2014 The Go Circuit Project
// Use of this source code is governed by the license for
// The Go Circuit Project, found in the LICENSE file.
//
// Authors:
//   2014 Chris Monson <chris@gocircuit.org>

/*
Implementation of a select statement using file system primitives.
*/

package ripple

import (
	"bytes"
	"encoding/json"
	"fmt"
	"io"
	"io/ioutil"
	"log"
	"math/rand"
	"os"
	"path/filepath"
	"strings"
	"sync"
	"syscall"
	"time"

	"circuit/kit/rh"
	"circuit/kit/rh/ns"
	"circuit/kit/rh/shargs"
)

type FileCase struct {
	Chan string `json:"chan"`
	Data string `json:"data"`
}

func (fc FileCase) String() string {
	return fmt.Sprintf("FileCase: chan=%q data=%q", fc.Chan, fc.Data)
}

func buildSelectHelp(name string, id uint64) string {
	return fmt.Sprintf(selectHelpFmt, filepath.Join("/", name), id)
}

const selectHelpFmt = `
  This is a select ripple, named "%s", with QID ⟨Q%016x⟩.

  SYNOPSIS

  cat help    # This help screen.

  cat <<EOF >> cases_json
  [{"chan": "mychan/recv"},
   {"chan": "mychan/send",
    "data": "a unicode string to send"}]
  EOF
  cat select  # Returns when a case is satisfied.
			  # Format is "<i>:<name>\n<data>" where
			  # <i> is the case index,
			  # <name> is the channel file name, and
			  # <data> is raw byte data from a receive channel, possibly empty.

  The select file blocks until one of the cases is satisifed. As with Go's
  select statement, specifying no cases results in blocking forever.

  A special channel called "default" can be added if the select should not
  block when no channels are available. Otherwise, all cases should specify a
  path ending in either "/recv" or "/send"

  # Quoted
  echo '"mychan/recv" "mychan/send" "stuff"' >> cases_quoted

  # Line-delimited
  cat <<EOF >> cases_lines
  mychan/recv
  mychan/send
  stuff with\nembedded newlines
  EOF

  Note that unicode is assumed in all cases. If you wish to send arbitrary
  bytes, you may choose to encode them in e.g., base64 or some other
  appropriate unicode-friendly encoding.

  If the select blocks and you want to unstick it, you can write anything to
  the "interrupt" endpoint, and it will immediately stop trying to select:

  echo >> interrupt

`

type Select struct {
	sync.Mutex

	q    rh.Q
	name string
	gtm  time.Time

	cases []FileCase
	intr  chan bool

	casesQuotedFID rh.FID
	casesLinesFID  rh.FID
	casesJsonFID   rh.FID
	interruptFID   rh.FID
	selectFID      rh.FID
	helpFID        rh.FID
}

func NewSelect(name string) (*Select, rh.FID) {
	now := time.Now()
	s := &Select{
		q: rh.Q{
			ID:  uint64(rand.Int63()),
			Ver: 1,
		},
		name: name,
		gtm:  now,
		intr: make(chan bool, 1),
	}

	s.casesQuotedFID = NewConfigFID(
		"cases_quoted",
		func(b []byte) error {
			parts, err := shargs.Split(string(b))
			if err != nil {
				return err
			}
			if len(parts) == 0 {
				return fmt.Errorf("Empty cases")
			}
			cases, err := stringsToCases(parts)
			if err != nil {
				return err
			}
			return s.SetCases(cases)
		},
		func() []byte {
			var strs []string
			for _, c := range s.Cases() {
				strs = append(strs, shargs.Quote(c.Chan))
				if strings.HasSuffix(c.Chan, "send") {
					strs = append(strs, shargs.Quote(c.Data))
				}
			}
			return []byte(strings.Join(strs, " "))
		},
	)

	s.casesLinesFID = NewConfigFID(
		"cases_lines",
		func(b []byte) error {
			r := strings.NewReplacer(
				`\0`, "\x00",
				`\n`, "\n",
				`\r`, "\r",
				`\\`, `\`,
			)
			// TODO(chris): use scanner instead to capture \r\n, etc.
			lines := strings.Split(string(b), "\n")
			for i, line := range lines {
				lines[i] = r.Replace(line)
			}
			cases, err := stringsToCases(lines)
			if err != nil {
				return err
			}
			return s.SetCases(cases)
		},
		func() []byte {
			r := strings.NewReplacer(
				`\`, `\\`,
				"\r", `\r`,
				"\n", `\n`,
				"\x00", `\0`,
			)
			var strs []string
			for _, c := range s.Cases() {
				strs = append(strs, c.Chan)
				if strings.HasSuffix(c.Chan, "send") {
					strs = append(strs, r.Replace(c.Data))
				}
			}
			return []byte(strings.Join(strs, "\n"))
		},
	)

	s.casesJsonFID = NewConfigFID(
		"cases_json",
		func(b []byte) error {
			var cases []FileCase
			err := json.Unmarshal(b, &cases)
			if err != nil {
				log.Printf("Error: Case JSON did not parse: %v", err)
				return err
			}
			cases, err = normalizeCases(cases)
			if err != nil {
				return err
			}
			return s.SetCases(cases)
		},
		func() []byte {
			cases := s.Cases()
			bytes, err := json.MarshalIndent(cases, "", "  ")
			if err != nil {
				log.Printf("Error marshalling cases %v: %v", cases, err)
				return nil
			}
			return bytes
		},
	)

	s.interruptFID = NewEndpoint(func(_ string) {
		s.intr <- true
	})

	s.selectFID = newSelectFID(s)
	s.helpFID = NewHelp(buildSelectHelp(name, uint64(rand.Int63())))

	return s, s.Walk()
}

func stringsToCases(strs []string) ([]FileCase, error) {
	var cases []FileCase
	for i := 0; i < len(strs); i++ {
		c := strs[i]
		if i == len(strs)-1 && c == "" {
			// Sometimes the last line will be a blank.
			break
		}
		curcase := FileCase{Chan: c}
		base := filepath.Base(c)
		if base == "send" || base == "trysend" {
			i++
			if i < len(strs) {
				curcase.Data = strs[i]
			} else {
				return nil, fmt.Errorf("Error: send channel without data: %v", c)
			}
		}
		cases = append(cases, curcase)
	}
	return normalizeCases(cases)
}

func normalizeCases(cases []FileCase) (normalized []FileCase, err error) {
	hasDefault := false
	for i, c := range cases {
		base := filepath.Base(c.Chan)
		dir := filepath.Dir(c.Chan)
		switch base {
		case "tryrecv", "recv":
			base = "tryrecv"
		case "trysend", "send":
			base = "trysend"
		case "default":
			if hasDefault {
				err = fmt.Errorf("Extra default case at case %d", i)
				return
			}
			if dir != "." {
				err = fmt.Errorf("Weird default has a path at case %d: %v", i, c.Chan)
				return
			}
			hasDefault = true
		default:
			err = fmt.Errorf(
				"Unknown channel direction %q specified at item %d: %v", base, i, c.Chan)
			return
		}
		chanpath := filepath.Join(dir, base)
		normalized = append(normalized, FileCase{Chan: chanpath, Data: c.Data})
	}
	return normalized, nil
}

func (s *Select) SetCases(cases []FileCase) error {
	defer un(lock(s))
	s.cases = make([]FileCase, len(cases))
	copy(s.cases, cases)
	return nil
}

func (s *Select) Cases() []FileCase {
	defer un(lock(s))
	cases := make([]FileCase, len(s.cases))
	copy(cases, s.cases)
	return cases
}

func (s *Select) Walk() rh.FID {
	return ns.NewFID("select", &selectDir{s})
}

func (s *Select) Read(p []byte) (n int, err error) {
	return 0, rh.ErrPerm
}

func (s *Select) Write(p []byte) (n int, err error) {
	return 0, rh.ErrPerm
}

// The "select" file within the ripple. Read this to block and select.
type selectFID struct {
	rh.AnchoredFID
	sync.Mutex

	q   rh.Q
	gtm time.Time

	sel *Select

	chosenCase FileCase
	recvReader *bytes.Reader
}

func newSelectFID(s *Select) rh.FID {
	return &selectFID{
		sel: s,
		gtm: time.Now(),
		q: rh.Q{
			ID:  uint64(rand.Int63()),
			Ver: 1,
		},
	}
}

func (s *selectFID) String() string {
	return fmt.Sprintf("select (%p) %v", s, s.Q())
}

func (s *selectFID) Q() rh.Q {
	return s.q
}

func (s *selectFID) attemptSend(fc FileCase) bool {
	file, err := os.OpenFile(fc.Chan, os.O_APPEND|os.O_WRONLY, os.ModeAppend|0222)
	if err != nil {
		if err.(*os.PathError).Err != syscall.EBUSY {
			log.Printf("Trysend failed for case %v: %v", fc, err)
		}
		return false
	}
	defer file.Close()
	data := []byte(fc.Data)
	n, err := file.Write(data)
	if n != len(data) {
		log.Printf("Only wrote %d out of %d bytes", n, len(data))
	}
	if err != nil {
		log.Printf("Error writing to channel %#v: %#v", fc.Chan, err)
		// Once open, we have selected this successfully, even if the write itself fails.
		return true
	}
	return true
}

func (s *selectFID) attemptRecv(fc FileCase) ([]byte, bool) {
	file, err := os.Open(fc.Chan)
	if err != nil {
		if err.(*os.PathError).Err != syscall.EBUSY {
			log.Printf("Tryrecv failed for case %v: %v", fc, err)
		}
		return nil, false
	}
	defer file.Close()
	all, err := ioutil.ReadAll(file)
	if err != nil {
		log.Printf("Error reading from channel %v: %v", fc.Chan, err)
	}
	return all, true
}

func (s *selectFID) attemptSelect() (chosen int, name string, data []byte) {
	defer un(lock(s.sel)) // protect s.sel.cases access
	indices := rand.Perm(len(s.sel.cases))
	defaultIndex := -1
	for _, i := range indices {
		c := s.sel.cases[i]
		switch dir, base := filepath.Dir(c.Chan), filepath.Base(c.Chan); base {
		case "tryrecv":
			chanpath := filepath.Join(dir, "tryrecv")
			data, ok := s.attemptRecv(FileCase{Chan: chanpath, Data: c.Data})
			if ok {
				return i, c.Chan, data
			}
		case "trysend":
			chanpath := filepath.Join(dir, "trysend")
			ok := s.attemptSend(FileCase{Chan: chanpath, Data: c.Data})
			if ok {
				return i, c.Chan, nil
			}
		case "default":
			defaultIndex = i
			// Don't do anything else with this unless we exhaust all others first.
		default:
			panic(fmt.Sprintf("Can't get here - normalized case with weird suffix: %v", c.Chan))
		}
	}
	if defaultIndex >= 0 {
		return defaultIndex, s.sel.cases[defaultIndex].Chan, nil
	}
	return -1, "", nil
}

func (s *selectFID) Open(flag rh.Flag, intr rh.Intr) (err error) {
	defer un(lock(s))

	// avoid blocking simulated opens from the fuse/rh setattr hack
	if flag.Create || flag.Truncate || flag.Deny {
		return rh.ErrPerm
	}

	// Loop through cases, randomized each time, and try to receive or send on
	// the channel as appropriate. Note that this loop executes exactly once if
	// there is a default case. Otherwise it shuffles and tries again.
	for {
		chosen, name, data := s.attemptSelect()
		if chosen >= 0 {
			all := []byte(fmt.Sprintf("%d:%s\n", chosen, name))
			all = append(all, data...)
			s.recvReader = bytes.NewReader(all)
			break
		}
		// Nothing was chosen, and no default. Sleep and try again.
		select {
		case <-intr:
			return rh.ErrGone
		case <-s.sel.intr:
			return rh.ErrGone
		case <-time.After(2 * time.Second):
			// Nothing - just go again.
		}
	}
	return nil
}

func (s *selectFID) Read(_ int64, count int, intr rh.Intr) (chunk rh.Chunk, err error) {
	defer un(lock(s))
	buff := make([]byte, count)
	n, err := s.recvReader.Read(buff)
	if err != nil && err == io.EOF {
		err = nil // fuse has a weird problem with EOF errors.
	}
	return rh.ByteChunk(buff[:n]), err
	return nil, nil
}

func (s *selectFID) Write(offset int64, data rh.Chunk, intr rh.Intr) (n int, err error) {
	defer un(lock(s))
	return 0, nil
}

func (s *selectFID) Clunk() {
}

func (s *selectFID) Stat() (*rh.Dir, error) {
	return &rh.Dir{
		Q:      s.q,
		Mode:   rh.Mode{Attr: rh.ModeIO},
		Perm:   0444, // -r--r--r-, acts as /dev/null for writing and empty file for reading
		Atime:  s.gtm,
		Mtime:  s.gtm,
		Name:   "select",
		Length: 0,
		Uid:    uid(),
		Gid:    gid(),
	}, nil
}

func (s *selectFID) Walk(wname []string) (rh.FID, error) {
	if len(wname) > 0 {
		return nil, rh.ErrClash
	}
	return newSelectFID(s.sel), nil
}

type selectDir struct {
	s *Select
}

func (sd *selectDir) Help() rh.FID {
	return sd.s.helpFID
}

func (sd *selectDir) Cases_quoted() rh.FID {
	return sd.s.casesQuotedFID
}

func (sd *selectDir) Cases_lines() rh.FID {
	return sd.s.casesLinesFID
}

func (sd *selectDir) Cases_json() rh.FID {
	return sd.s.casesJsonFID
}

func (sd *selectDir) Select() rh.FID {
	return sd.s.selectFID
}

func (sd *selectDir) Interrupt() rh.FID {
	return sd.s.interruptFID
}
